package com.shlh.saas.util;

import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 线程池工厂类
 * 用于创建和管理应用程序中的线程池，提升数据处理性能
 */
@Component
@Slf4j
public class ThreadPoolFactory {

    /**
     * 数据查询专用线程池
     */
    private ThreadPoolExecutor dataQueryExecutor;

    /**
     * 数据处理专用线程池
     */
    private ThreadPoolExecutor dataProcessExecutor;

    /**
     * 导出专用线程池
     */
    private ThreadPoolExecutor exportExecutor;

    // 线程池配置参数
    private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();
    private static final int CORE_POOL_SIZE = Math.max(2, CPU_COUNT / 2);
    private static final int MAX_POOL_SIZE = CPU_COUNT * 2;
    private static final long KEEP_ALIVE_TIME = 60L;
    private static final int QUEUE_CAPACITY = 100;

    /**
     * 自定义线程工厂
     */
    private static class CustomThreadFactory implements ThreadFactory {
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String namePrefix;

        CustomThreadFactory(String namePrefix) {
            this.namePrefix = namePrefix;
        }

        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r, namePrefix + "-thread-" + threadNumber.getAndIncrement());
            if (t.isDaemon()) {
                t.setDaemon(false);
            }
            if (t.getPriority() != Thread.NORM_PRIORITY) {
                t.setPriority(Thread.NORM_PRIORITY);
            }
            return t;
        }
    }

    /**
     * 自定义拒绝策略
     */
    private static class CustomRejectedExecutionHandler implements RejectedExecutionHandler {
        private final String poolName;

        CustomRejectedExecutionHandler(String poolName) {
            this.poolName = poolName;
        }

        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            log.warn("线程池 {} 任务被拒绝执行，当前活跃线程数: {}, 队列大小: {}, 任务总数: {}",
                    poolName, executor.getActiveCount(), executor.getQueue().size(), executor.getTaskCount());
            // 使用调用者线程执行任务
            if (!executor.isShutdown()) {
                r.run();
            }
        }
    }

    /**
     * 初始化线程池
     */
    @PostConstruct
    public void init() {
        log.info("初始化线程池工厂，CPU核心数: {}", CPU_COUNT);

        // 数据查询线程池
        dataQueryExecutor = new ThreadPoolExecutor(
                CORE_POOL_SIZE,
                MAX_POOL_SIZE,
                KEEP_ALIVE_TIME,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(QUEUE_CAPACITY),
                new CustomThreadFactory("data-query"),
                new CustomRejectedExecutionHandler("data-query")
        );

        // 数据处理线程池
        dataProcessExecutor = new ThreadPoolExecutor(
                CORE_POOL_SIZE,
                MAX_POOL_SIZE,
                KEEP_ALIVE_TIME,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(QUEUE_CAPACITY),
                new CustomThreadFactory("data-process"),
                new CustomRejectedExecutionHandler("data-process")
        );

        // 导出专用线程池（较大的队列容量）
        exportExecutor = new ThreadPoolExecutor(
                CORE_POOL_SIZE,
                MAX_POOL_SIZE,
                KEEP_ALIVE_TIME,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(QUEUE_CAPACITY * 2),
                new CustomThreadFactory("export"),
                new CustomRejectedExecutionHandler("export")
        );

        log.info("线程池初始化完成 - 数据查询池: {}/{}, 数据处理池: {}/{}, 导出池: {}/{}",
                dataQueryExecutor.getCorePoolSize(), dataQueryExecutor.getMaximumPoolSize(),
                dataProcessExecutor.getCorePoolSize(), dataProcessExecutor.getMaximumPoolSize(),
                exportExecutor.getCorePoolSize(), exportExecutor.getMaximumPoolSize());
    }

    /**
     * 获取数据查询线程池
     */
    public ThreadPoolExecutor getDataQueryExecutor() {
        return dataQueryExecutor;
    }

    /**
     * 获取数据处理线程池
     */
    public ThreadPoolExecutor getDataProcessExecutor() {
        return dataProcessExecutor;
    }

    /**
     * 获取导出线程池
     */
    public ThreadPoolExecutor getExportExecutor() {
        return exportExecutor;
    }

    /**
     * 提交数据查询任务
     */
    public <T> Future<T> submitDataQuery(Callable<T> task) {
        logPoolStatus("data-query", dataQueryExecutor);
        return dataQueryExecutor.submit(task);
    }

    /**
     * 提交数据处理任务
     */
    public <T> Future<T> submitDataProcess(Callable<T> task) {
        logPoolStatus("data-process", dataProcessExecutor);
        return dataProcessExecutor.submit(task);
    }

    /**
     * 提交导出任务
     */
    public <T> Future<T> submitExport(Callable<T> task) {
        logPoolStatus("export", exportExecutor);
        return exportExecutor.submit(task);
    }

    /**
     * 记录线程池状态
     */
    private void logPoolStatus(String poolName, ThreadPoolExecutor executor) {
        if (log.isDebugEnabled()) {
            log.debug("线程池 {} 状态 - 活跃线程: {}, 队列大小: {}, 完成任务数: {}, 总任务数: {}",
                    poolName,
                    executor.getActiveCount(),
                    executor.getQueue().size(),
                    executor.getCompletedTaskCount(),
                    executor.getTaskCount());
        }
    }

    /**
     * 获取线程池监控信息
     */
    public String getPoolMonitorInfo() {
        StringBuilder sb = new StringBuilder();
        sb.append("=== 线程池监控信息 ===\n");
        
        appendPoolInfo(sb, "数据查询池", dataQueryExecutor);
        appendPoolInfo(sb, "数据处理池", dataProcessExecutor);
        appendPoolInfo(sb, "导出池", exportExecutor);
        
        return sb.toString();
    }

    private void appendPoolInfo(StringBuilder sb, String poolName, ThreadPoolExecutor executor) {
        sb.append(String.format("%s: 核心线程=%d, 最大线程=%d, 活跃线程=%d, 队列大小=%d, 完成任务=%d\n",
                poolName,
                executor.getCorePoolSize(),
                executor.getMaximumPoolSize(),
                executor.getActiveCount(),
                executor.getQueue().size(),
                executor.getCompletedTaskCount()));
    }

    /**
     * 优雅关闭线程池
     */
    @PreDestroy
    public void shutdown() {
        log.info("开始关闭线程池...");
        
        shutdownExecutor("data-query", dataQueryExecutor);
        shutdownExecutor("data-process", dataProcessExecutor);
        shutdownExecutor("export", exportExecutor);
        
        log.info("所有线程池已关闭");
    }

    private void shutdownExecutor(String poolName, ThreadPoolExecutor executor) {
        if (executor != null && !executor.isShutdown()) {
            log.info("关闭线程池: {}", poolName);
            executor.shutdown();
            try {
                if (!executor.awaitTermination(30, TimeUnit.SECONDS)) {
                    log.warn("线程池 {} 在30秒内未能正常关闭，强制关闭", poolName);
                    executor.shutdownNow();
                    if (!executor.awaitTermination(10, TimeUnit.SECONDS)) {
                        log.error("线程池 {} 无法强制关闭", poolName);
                    }
                }
            } catch (InterruptedException e) {
                log.error("等待线程池 {} 关闭时被中断", poolName, e);
                executor.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }
    }

    /**
     * 等待所有任务完成
     */
    public void awaitAllTasks(long timeout, TimeUnit unit) throws InterruptedException {
        long startTime = System.currentTimeMillis();
        long timeoutMillis = unit.toMillis(timeout);
        
        while (hasActiveTasks() && (System.currentTimeMillis() - startTime) < timeoutMillis) {
            Thread.sleep(100);
        }
        
        if (hasActiveTasks()) {
            log.warn("等待任务完成超时，仍有活跃任务");
        }
    }

    private boolean hasActiveTasks() {
        return (dataQueryExecutor.getActiveCount() > 0) ||
               (dataProcessExecutor.getActiveCount() > 0) ||
               (exportExecutor.getActiveCount() > 0);
    }
}
